package com.niodata.md.rdb.crawler;

import com.niodata.md.rdb.model.Catalog;
import com.niodata.md.rdb.model.DataBase;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.commons.lang3.ArrayUtils;

/**
 * rdb metadata loader.
 */
public class RdbMetaDataLoader {

  /**
   * list all catalog names.
   *
   * @return all catalog
   */
  public List<String> listCatalogs(DataSourceInfo dataSourceInfo, DataBaseRetrievalOptions options) throws Exception {
    Connection connection = dataSourceInfo.getConnection();
    try {
      DatabaseMetaData metaData = connection.getMetaData();
      ResultSet rs = metaData.getCatalogs();
      String regx = options.namePattern();
      if ("%".equals(regx)) {
        regx = ".*";
      }
      List<String> names = new ArrayList<>();
      while (rs.next()) {
        String catalogName = rs.getString(1);
        if (catalogName.matches(regx) && !ArrayUtils.contains(options.nameExclusions(), catalogName)) {
          names.add(catalogName);
        }
      }
      return names;
    } catch (Exception ex) {
      ex.printStackTrace();
    } finally {
      connection.close();
    }
    return null;
  }

  /**
   * load catalog.
   *
   * @param catalogName catalog name
   * @return catalog
   */
  public Catalog loadCatalog(String catalogName,
                             DataSourceInfo dataSourceInfo,
                             DataBaseRetrievalOptions dataBaseRetrievalOptions) throws Exception {
    Connection connection = dataSourceInfo.getConnection(catalogName);
    Catalog catalog = new Catalog();
    catalog.setName(catalogName);
    try {
      List<DataBase> dataBaseList = new ArrayList<>();
      DataBase dataBase = new DataBase();
      dataBase.setCatalog(catalogName);
      dataBase.setName(catalogName);
      dataBaseList.add(dataBase);
      catalog.setDataBaseList(dataBaseList);
      LinkedBlockingQueue<MetaDataRetrieval> retrievals = new LinkedBlockingQueue<>();
      DbInfoRetrieval dbInfoRetrieval = new DbInfoRetrieval();
      retrievals.add(dbInfoRetrieval);
      dbInfoRetrieval.setCatalog(catalog);
      dbInfoRetrieval.setMetaData(connection.getMetaData());
      dbInfoRetrieval.setDataSourceInfo(dataSourceInfo);
      String[] parentPath = new String[]{catalogName, catalogName};
      dbInfoRetrieval.setParentPath(parentPath);

      TableRetrieval tableRetrieval = new TableRetrieval();
      tableRetrieval.withTableRetrievalOptions(dataBaseRetrievalOptions.getTableRetrievalOptions());
      retrievals.add(tableRetrieval);
      tableRetrieval.setCatalog(catalog);
      tableRetrieval.setMetaData(connection.getMetaData());
      tableRetrieval.setDataSourceInfo(dataSourceInfo);
      tableRetrieval.setParentPath(parentPath);

      while (retrievals.peek() != null) {
        MetaDataRetrieval retrieval = retrievals.poll();
        List<MetaDataRetrieval> subRetrievals = retrieval.retrieveMetaData();
        if (subRetrievals.size() > 0) {
          retrievals.addAll(subRetrievals);
        }
      }
    } finally {
      connection.close();
    }
    return catalog;
  }
}
